package com.yeyks.app.config;

import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.PropertyValueConst;
import com.yeyks.app.config.properties.RocketMQProperty;
import com.yeyks.app.function.rocketmq.consumer.GrabTheOrderSearchMessageListener;
import com.yeyks.app.function.rocketmq.consumer.GrabTheOrderWebsocketMessageListener;
import com.yeyks.app.function.rocketmq.consumer.WebToCSendMessageListener;
import com.yeyks.common.constants.rocketmq.TopicAndTag;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.Properties;

@Slf4j
@Component
@ConditionalOnProperty(prefix = "rocket-mq", name = "enable", havingValue = "true")
public class RocketMQConsumerConfig {

    @Autowired
    private RocketMQProperty rocketMQProperty;

    //抢单监听
    @Bean(name = "grabTheOrderSearchConsumer", initMethod = "start", destroyMethod = "shutdown")
    public Consumer grabTheOrderSearchConsumer(GrabTheOrderSearchMessageListener grabTheOrderSearchMessageListener) {
        Properties properties = getRocketMQBaseProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_YYKSNORMAL_WEBTOCTAG);
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_WEB_TO_C_TAG, grabTheOrderSearchMessageListener);
        return consumer;
    }

    //抢单websocket推送监听
    @Bean(name = "grabTheOrderWebsocketConsumer", initMethod = "start", destroyMethod = "shutdown")
    public Consumer grabTheOrderWebsocketConsumer(GrabTheOrderWebsocketMessageListener grabTheOrderWebsocketMessageListener) {
        Properties properties = getRocketMQBaseProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_YYKSNORMAL_GRABTHEORDERWEBSOCKET);
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_GRAB_THE_ORDER_WEBSOCKET, grabTheOrderWebsocketMessageListener);
        return consumer;
    }

    /**
     * 消息推送（例如活动推送）
     *
     * @return
     */
    @Bean(name = "webToCSendConsumer")
    public Consumer webToCSendConsumer(WebToCSendMessageListener webToCSendMessageListener) {
        log.info("初始化webToCSendConsumer启动消费者！");
        Properties properties = getRocketMQBaseProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_YYKSDELAY_WEBTOCSENDTAG);
        // 集群订阅方式
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_WEB_TO_C_SEND_TAG, webToCSendMessageListener);
        consumer.start();
        log.info("启动webToCSendConsumer消费者成功！");
        return consumer;
    }

    //rocket mq 基础消费者属性
    public Properties getRocketMQBaseProperties() {
        Properties properties = new Properties();
        properties.setProperty(PropertyKeyConst.AccessKey, rocketMQProperty.getAccessKey());
        properties.setProperty(PropertyKeyConst.SecretKey, rocketMQProperty.getSecretKey());
        properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, rocketMQProperty.getNameSrvAddr());
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "10");
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
        return properties;
    }

}
